/**************************************************************************************************************
 * @copyright Copyright (c) 2015—2024 Uniorange Internet Design Co., LTD. All rights reserved.
 * @author Uniorange Internet Design Co., LTD.
 *
 * This file is part of the App project. Copyright © Uniorange Internet Design Co., Ltd. All Rights Reserved.
 * Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential.
 *
 **************************************************************************************************************/
package com.jinyu.gateway.handler;

import com.fasterxml.jackson.databind.util.BeanUtil;
import com.jinyu.core.protocol.*;
import com.jinyu.gateway.registry.RegistryCenter;
import com.jinyu.gateway.registry.ServerNode;
import com.jinyu.gateway.registry.SubscribeCenter;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 邱金波
 * @version V0.1
 * @className RegisterProtocolHandler
 * @description 注册协议处理器
 * @date 2024/09/18 19:37
 */
@ChannelHandler.Sharable
@Component
@Slf4j
public class ProtocolDispatchHandler extends SimpleChannelInboundHandler<Header> {
    private static AtomicInteger requestTimes = new AtomicInteger(0);
    private static ScheduledExecutorService executor = Executors.newScheduledThreadPool(1, r -> {
        return new Thread(r, "qps-counter");
    });

    static {
        executor.scheduleAtFixedRate(() -> {
            log.info("qps:{}", requestTimes.getAndSet(0));
        }, 0, 1000, TimeUnit.MILLISECONDS);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Header header) throws Exception {
        requestTimes.incrementAndGet();
        if (header instanceof RegisterProtocol) {
            RegisterProtocol registerProtocol = (RegisterProtocol) header;
            RegistryCenter.register(registerProtocol.getName(), registerProtocol.getHost(), registerProtocol.getPort(), channelHandlerContext.channel());
            BoolProtocol boolProtocol = new BoolProtocol(Boolean.TRUE);
            channelHandlerContext.writeAndFlush(boolProtocol);
        } else if (header instanceof SubscribeProtocol) {
            SubscribeProtocol subscribeProtocol = (SubscribeProtocol) header;
            String[] subscribeArrays = subscribeProtocol.getSubscribeArrays();
            SubscribeCenter.subscribe(subscribeProtocol.getConsumerName(), List.of(subscribeArrays));
            BoolProtocol boolProtocol = new BoolProtocol(Boolean.TRUE);
            channelHandlerContext.writeAndFlush(boolProtocol);
        } else if (header instanceof PingProtocol) {
            PongProtocol pongProtocol = new PongProtocol();
            pongProtocol.setType(ProtocolTypeEnum.PONG.getType());
            pongProtocol.setVersion(1);
            channelHandlerContext.writeAndFlush(pongProtocol);
        } else if (header instanceof UnSubscribeProtocol) {
            UnSubscribeProtocol unSubscribeProtocol = (UnSubscribeProtocol) header;
            SubscribeCenter.unSubscribe(unSubscribeProtocol.getName(), unSubscribeProtocol.getSubscribeList());
            BoolProtocol boolProtocol = new BoolProtocol(Boolean.TRUE);
            channelHandlerContext.writeAndFlush(boolProtocol);
        } else if (header instanceof GetSubscribeList) {
            GetSubscribeList getSubscribeList = (GetSubscribeList) header;
            Set<ServerNode> notifyList = SubscribeCenter.getNotifyList(getSubscribeList.getData());
            SubscribeRetProtocol subscribeRetProtocol = new SubscribeRetProtocol();
            List<com.jinyu.core.model.ServerNode> serverNodeList = new ArrayList<>();
            for (ServerNode serverNode : notifyList) {
                com.jinyu.core.model.ServerNode serverNode1 = new com.jinyu.core.model.ServerNode();
                BeanUtils.copyProperties(serverNode, serverNode1);
                serverNodeList.add(serverNode1);
            }
            subscribeRetProtocol.setData(serverNodeList);
        } else {
            //TODO: 暂不支持其他协议
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.WRITER_IDLE) {
                //如果一定时间内，没有写事件，就从注册中心中移除该连接
                String name = RegistryCenter.unRegister(ctx.channel().id());
                SubscribeCenter.unSubscribeByName(name);
            } else {
                log.info("DON't CARE the event");
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("exceptionCaught:{}, {}", ctx.channel().id(), cause);
        String name = RegistryCenter.unRegister(ctx.channel().id());
        SubscribeCenter.unSubscribeByName(name);
        ctx.close();
    }
}
